import logging
from multiprocessing import Process, Pipe
from threading import RLock
import sys
import types


_builtin_function_or_method = type(sum)


class ProcessPool:
    def __init__(self, size):
        self.size = size
        self.processes = {}
        self.lock = RLock()
        self.stopping = False

    def call_asynch(self, func, *args, **kwargs):
        self._call(False, func, *args, **kwargs)

    def call_synch(self, func, *args, **kwargs):
        return self._call(True, func, *args, **kwargs)

    def _call(self, synch, func, *args, **kwargs):
        with self.lock:
            found = False
            for v in self.processes.values():
                if not v[1]:
                    v[1] = True
                    found = True
                    break
            if not found:
                sender, receiver = Pipe(True)
                p = Process(target=run, args=(receiver,), daemon=True)
                v = [sender, True]
                self.processes[p] = v
                p.start()
                logging.getLogger().info('use %d processes' % len(self.processes))

        try:
            v[0].send((synch, func, args, kwargs))
            if synch:
                result = v[0].recv()
                if isinstance(result, Exception):
                    raise result
                return result
        finally:
            with self.lock:
                v[1] = False

    def start(self):
        logging.getLogger().info('start process pool')
        with self.lock:
            self.stopping = False
            for _ in range(self.size):
                sender, receiver = Pipe(True)
                self.processes[Process(target=run, args=(receiver,), daemon=True)] = [sender, False]
            for p in self.processes:
                p.start()
        logging.getLogger().info('Process pool is started')

    def stop(self):
        with self.lock:
            self.stopping = True
            for p, v in self.processes.items():
                v[0].send('Stop')
                p.join()
            self.processes.clear()


def run(pipe):
    logger = logging.getLogger()
    logger.addHandler(logging.StreamHandler(sys.stdout))
    while True:
        try:
            value = pipe.recv()
            if isinstance(value, str):
                return
            synch, func, args, kwargs = value
            try:
                result = func(*args, **kwargs)
            except Exception as e:
                result = e
            finally:
                if synch:
                    pipe.send(result)
        except Exception as e:
            logger.error(e)


class RemoteObj:
    def __init__(self, cls, d:dict):
        self.cls = cls
        self.d = d


class RemoteRef:
    def __init__(self, names):
        self.names = names


def obj2remote(obj, names=None, store: dict = None):
    if obj is None:
        return obj

    _id = id(obj)
    if store is None:
        store = {}
        names = []
    else:
        if _id in store:
            return store[_id]

    store[_id] = RemoteRef(names)
    t = type(obj)
    if t in (int, float, str, bool):
        return obj
    if t == tuple:
        return tuple([obj2remote(e, names, store) for e in obj])
    if t == list:
        return [obj2remote(e, names, store) for e in obj]
    if t == dict:
        return {obj2remote(k, names, store): obj2remote(v, names, store) for k, v in obj.items()}

    d = {}
    for attr in dir(obj):
        if not (attr.startswith('__') and attr.endswith('__')):
            v = getattr(obj, attr)
            if type(v) not in (types.FunctionType, types.MethodType, _builtin_function_or_method):
                new_names = names.copy()
                new_names.append(attr)
                d[attr] = obj2remote(v, new_names, store)
    return RemoteObj(obj.__class__, d)


def remote2obj(obj, names=None, root=None):
    if obj is None:
        return obj

    if names is None:
        names = []
        root = None

    t = type(obj)
    if t in (int, float, str, bool):
        return obj
    if t == tuple:
        return tuple([remote2obj(e, names, root) for e in obj])
    if t == list:
        return [remote2obj(e, names, root) for e in obj]
    if t == dict:
        return {remote2obj(k, names, root): remote2obj(v, names, root) for k, v in obj.items()}

    if isinstance(obj, RemoteObj):
        d = obj.d
        obj = obj.cls()
        if root is None:
            root = obj
        for name, value in d.items():
            new_names = names.copy()
            new_names.append(name)
            setattr(obj, name, remote2obj(value, names, root))
        return obj
    if isinstance(obj, RemoteRef):
        return get_obj_by_names(root, obj.names)

    raise AssertionError('Can not convert the remote object %s into a local object' % obj)


def get_obj_by_names(obj, names):
    if len(names) == 0:
        return obj

    obj = getattr(obj, names[0])
    return get_obj_by_names(obj, names[1:])


def _f(idx):
    print(idx, 'begin')
    n = 1.0
    for i in range(10000000):
        n *= 0.99999999999
    print(idx, 'end')
    return n


if __name__ == '__main__':
    size = 3
    from threading import Thread
    import time

    print('testing threads')
    ths = [Thread(target=_f, args=(i,), daemon=True) for i in range(size)]
    t = time.time()
    for th in ths:
        th.start()
    for th in ths:
        th.join()
    t_th = time.time() - t
    print('t_th: ', t_th)

    pool = ProcessPool(size)
    pool.start()
    time.sleep(3)
    print('testing processes')

    ths = [Thread(target=pool.call, args=(_f, i), daemon=True) for i in range(size)]
    t = time.time()
    for th in ths:
        th.start()
    for th in ths:
        th.join()
    t_p = time.time() - t

    print('t_th: %.6f, t_p: %.6f, %.2f%%' % (t_th, t_p, (t_th - t_p)/t_th*100))
    pool.stop()
    print('Stopped')
    pool.start()
    pool.stop()
